<?php
namespace app\admin\controller;

use Workerman\Worker;
use Workerman\Lib\Timer;


require_once dirname(__DIR__) . "/../vendor/workerman/workerman/Autoloader.php";
require_once dirname(__DIR__) . "/../vendor/mysql-master/src/Connection.php";
require_once dirname(__DIR__) . "/../vendor/channel-master/src/Server.php";
require_once dirname(__DIR__) . "/../vendor/channel-master/src/Client.php";

define('HEARTBEAT_TIME', 55);

class Workerman
{
    private $worker = null;
    private $db = null;
    private $channel_server = null;
    private $redis = null;
    public function run()
    {
        $this->channel_server = new \Channel\Server('0.0.0.0', 11112);
        $this->worker = new Worker("websocket://0.0.0.0:11111");
        $this->worker->count = 4;
        $this->redis = new \Redis();
        $this->redis->connect('127.0.0.1', 6379);
        
        $this->worker->onWorkerStart = function ($worker) {
            $this->db = new \Workerman\MySQL\Connection('localhost', 3306, 'websocket', 'ccwkmzEKjDLkhdtW', 'websocket');
            if ($worker->id === 0) {
                Timer::add(0.1, function () {
                    //当redis isGetData存在是才去更新数据
                    if($this->redis->get('isGetData')){
                        $this->updateDataBaseS();
                    }
                });
                echo 'updateDataBase started';
            }
            Timer::add(1, function () {
                $this->heartBeat();
            });
            \Channel\Client::connect('127.0.0.1', 11112);
            // 订阅upDateDataBase事件，并注册事件回调
            \Channel\Client::on('upDateDataBase', function ($upData) {
                // 向当前worker进程的所有客户端广播消息
                foreach ($this->worker->connections as $connection) {
                    $json = json_encode($upData);
                    $connection->send($json);
                }
            });
        };
        $this->worker->onConnect = function ($connection) {
            //当有链接时开启redis 60s
            $this->redis->setex('isGetData', 60, true);
            $data = $this->db->select('*')->from('product')->query();
            $json = json_encode($data);
            $connection->send($json);
        };

        $this->worker->onMessage = function ($connection, $msg) {
            //心跳时间更新
            if ($msg == 'heartBeat') {
                $connection->lastMessageTime = time();
            }
        };

        Worker::runAll();
    }

    //新浪的接口
    private function updateDataBase()
    {
        $symbols = $this->db->select('id, symbol')->from('product')->query();
        if ($symbols) {
            $time = time();
            $list = '';
            foreach ($symbols as $k => $symbol) {
                $list .= ($symbol['symbol'] . ',');
            }
            $url = 'http://hq.sinajs.cn/rn=' . $time . 'list=' . $list;
            $getdata = $this->curlfun($url);
            $datas_arr = explode(';', $getdata);
            unset($datas_arr[count($datas_arr) - 1]);
            foreach ($datas_arr as $k => $v) {
                $data_arr = explode(',', $v);
                foreach ($symbols as $key => $val) {
                    if (strpos($data_arr[0], $val['symbol'])) {
                        $data_arr[0] = $val['id'];
                        $upData = [
                            'symbol' => $val['symbol'],
                            'price' => $data_arr[1],
                            'open' => $data_arr[5],
                            'close' => $data_arr[3],
                            'high' => $data_arr[6],
                            'low' => $data_arr[7],
                            'diff' => $data_arr[12],
                            'diff_rate' => $data_arr[4] / 10000,
                        ];
                        $res = $this->db->update('product')->cols($upData)->where('id=' . $data_arr[0])->query();
                        if ($res) {
                            $upData['id'] = $data_arr[0];
                            \Channel\Client::publish('upDateDataBase', $upData);
                        }

                        break;
                    }
                }
            }
        }
    }

    //阿里云接口
    private function upDateDataBaseS()
    {
        $host = "http://alirm-com.konpn.com/query/comrms";
        $querys = "?symbols=";
        $symbols = $this->db->select('id, symbol')->from('product')->query();
        foreach ($symbols as $symbol) {
            $querys .= ($symbol['symbol'] . ',');
        }
        $url = $host . $querys;
        $fulldata = json_decode($this->curlfun($url), true);
        if ($fulldata['Code'] === 0) {
            $getdata = $fulldata['Obj'];
            foreach ($symbols as $k => $symbol) {
                foreach ($getdata as $data) {
                    if ($symbol['symbol'] == $data['FS']) {
                        $upData = [
                            'symbol' => $data['FS'],
                            'price' => $data['P'],
                            'open' => $data['O'],
                            'close' => $data['YC'],
                            'high' => $data['H'],
                            'low' => $data['L'],
                            'zf' => $data['ZF'],
                            'vf' => $data['VF'],
                        ];
                        $res = $this->db->update('product')->cols($upData)->where('id=' . $symbol['id'])->query();
                        //当数据更新成功，通知其他connection广播信息
                        if ($res) {
                            $upData['id'] = $symbol['id'];
                            \Channel\Client::publish('upDateDataBase', $upData);
                        }
                        break;
                    }
                }
            }
        }
    }

    //心跳
    private function heartBeat()
    {
        $time_now = time();
        foreach ($this->worker->connections as $connection) {
            // 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
            if (empty($connection->lastMessageTime)) {
                $connection->lastMessageTime = $time_now;
                continue;
            }
            // 心跳过期，关闭客户连接
            if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
                $connection->close();
            }
        }
    }

    //curl获取数据
    private function curlfun($url, $params = array(), $method = 'GET')
    {
        $appcode = "";//填你自己的code
        $header = array();
        array_push($header, "Authorization:APPCODE " . $appcode);
        $opts = array(CURLOPT_TIMEOUT => 10, CURLOPT_RETURNTRANSFER => 1, CURLOPT_SSL_VERIFYPEER => false, CURLOPT_SSL_VERIFYHOST => false, CURLOPT_HTTPHEADER => $header);
        /* 根据请求类型设置特定参数 */
        switch (strtoupper($method)) {
            case 'GET':
                $opts[CURLOPT_URL] = $url . '?' . http_build_query($params);
                $opts[CURLOPT_URL] = substr($opts[CURLOPT_URL], 0, -1);

                break;
            case 'POST':
                //判断是否传输文件
                $params = http_build_query($params);
                $opts[CURLOPT_URL] = $url;
                $opts[CURLOPT_POST] = 1;
                $opts[CURLOPT_POSTFIELDS] = $params;
                break;
            default:
        }
        /* 初始化并执行curl请求 */
        $ch = curl_init();
        curl_setopt_array($ch, $opts);
        $data = curl_exec($ch);
        $error = curl_error($ch);
        curl_close($ch);
        if ($error) {
            $data = null;
        }
        return $data;
    }
}


$workerman = new workerman();
$workerman->run();
